package com.palfund.rxjava;

import android.os.Bundle;
import android.os.SystemClock;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.widget.Toast;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.Single;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Function3;
import io.reactivex.functions.Predicate;
import io.reactivex.schedulers.Schedulers;

public class MainActivity extends AppCompatActivity {


    private Disposable mDisposable;
    private Disposable mDispos;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        // 基本用法
        //baseMethod();
        //zip()方法将多个发射器发送的事件(数据)进行平行匹配,最终匹配成功的发射事件(数据)数目等于包含最少发送事件的发射器的发送事件的数目
        //zip();
        //concat()方法把2个发射器合并成1个发射器
        //concat();
        //merge()方法,把多个Observable结合起来.注意它和concat的区别在于,不是按照添加顺序连接,按照时间线连接,不用等到发射器A发送完所有的事件再进行发射器B的发送
        //merge();
        //timer()方法,延迟发送事件
        //timer();
        //interval()方法,用于间隔时间执行某个操作,其接受三个参数,分别是第一次发送延迟、间隔时间、时间单位
        //interval();
        //single()方法,只会接收一个参数,而SingleObserver只会调用onError()或者onSuccess()
        //single();
        //defer()方法,观察者订阅时才创建Observable,每次订阅返回一个新的Observable
        //defer();
        //range()方法,创建一个发射特定整数序列的Observable:第一个参数为起始值;第二个为发送的个数,如果为0则不发送,负数则抛异常.
        //range();
        //empty()方法,创建一个直接通知完成的Observable,直接调用onComplete()
        //empty();
        //error()方法,创建一个直接通知错误的Observable,直接调用onError()
        //error();
        //never()方法,创建一个什么都不做的Observable,onNext()、onComplete()、onError()均不调用
        //never();
        // 线程切换
        //changeThread();
        //map()方法促使发送的每一个事件都按照指定的函数去变化
        //map();
        //flatMap()平行发送事件,发射器将事件一起发出,可以把一个发射器Observable转换为多个Observables,
        // 然后再把这些分散的Observables装进一个单一的发射器Observable
        // .但有个需要注意的是,flatMap并不能保证事件的顺序
        //flatMap();
        //concatMap()顺序发送事件,上一个事件接收器接收后,concatMap发射器再发送下一个事件
        //concatMap();
        //distinct()方法,去除重复的事件
        //distinct();
        //filter()方法,根据特定的条件过滤掉不符合条件的事件
        //filter();
        //buffer()方法,将事件打包放在List集合中发送,参数count打包的事件数目,参数skip发送事件的步长,count>skip时部分事件会被重复打包发送,
        // count<skip时部分事件不会被打包发送
        //buffer();
        //doOnNext()方法,让订阅者在接收到数据之前干点有意思的事情
        //doOnNext();
        //skip()方法,代表跳过 count 个数目开始接收
        //skip();
        //take()接受一个long型参数count,代表至多接收count个数据。
        //take();
        //debounce()方法,去除与下一个事件发送时间间隔短的
        //debounce();
        //last()方法,当ObservableSource为空时发送last(defaultItem),不为空时发送可观察到的最后一个值
        //last();
        //reduce()方法,将发射器发送的事件依次进行处理,只发送最后的处理结果
        //reduce();
        //scan()方法,将发射器发送的事件依次进行处理,发送每次处理后的结果
        //scan();
        //window();
        //repeat();
        //repeat();
        //支持背压
        //Flowable.just(1).subscribe(new Subscriber<Integer>() {
        //            @Override
        //            public void onSubscribe(Subscription s) {
        //            }
        //
        //            @Override
        //            public void onNext(Integer integer) {
        //
        //            }
        //
        //            @Override
        //            public void onError(Throwable t) {
        //
        //            }
        //
        //            @Override
        //            public void onComplete() {
        //
        //            }
        //        });
        //        RxBus.getInstance().toObservable(Integer.class).subscribe(new Consumer<Integer>
        // () {
        //            public void accept(Integer integer) throws Exception {
        //                Log.i("MainActivity", "---accept()--->" + integer);
        //            }
        //        });
        //        RxBus.getInstance().post(1);
        //combineLatest()方法;发射器发送的数据与另一个发射器发送过的最后一个数据结合成新的数据
        //combineLatest();
        ArrayList<ArrayList<Integer>> lists = new ArrayList<>();
    }

    private void combineLatest() {
        Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                Log.i("MainActivity", "---subscribe()--->" + 1);
                e.onNext(1);
                SystemClock.sleep(5000);
                Log.i("MainActivity", "---subscribe()--->" + 2);
                e.onNext(2);
                SystemClock.sleep(10000);
                Log.i("MainActivity", "---subscribe()--->" + 3);
                e.onNext(3);
                SystemClock.sleep(3000);
                Log.i("MainActivity", "---subscribe()--->" + 4);
                e.onNext(4);
                SystemClock.sleep(5000);
                Log.i("MainActivity", "---subscribe()--->" + 5);
                e.onNext(5);
                e.onComplete();
            }
        }).subscribeOn(Schedulers.computation());

        Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                SystemClock.sleep(3000);
                Log.i("MainActivity", "---subscribe()--->" + "A");
                e.onNext("A");
                SystemClock.sleep(3000);
                Log.i("MainActivity", "---subscribe()--->" + "B");
                e.onNext("B");
                SystemClock.sleep(5000);
                Log.i("MainActivity", "---subscribe()--->" + "C");
                e.onNext("C");
                Log.i("MainActivity", "---subscribe()--->" + "D");
                e.onNext("D");
                e.onComplete();
            }
        }).subscribeOn(Schedulers.computation());
        //combineLatest()发送过的最后一个,
        Observable.combineLatest(observable1, observable2, new BiFunction<Integer, String,
                String>() {
            @Override
            public String apply(Integer integer, String s) throws Exception {
                return s + integer;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.i("MainActivity", "---accept()--->" + s);
            }
        });
    }

    private void never() {
        Observable.never().subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i("MainActivity", "---onSubscribe()--->");
            }

            @Override
            public void onNext(Object o) {
                Log.i("MainActivity", "---onNext()--->");
            }

            @Override
            public void onError(Throwable e) {
                Log.i("MainActivity", "---onError()--->");
            }

            @Override
            public void onComplete() {
                Log.i("MainActivity", "---onComplete()--->");
            }
        });
    }

    private void error() {
        Observable.error(new IllegalArgumentException("参数错误")).subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i("MainActivity", "---onSubscribe()--->");
            }

            @Override
            public void onNext(Object o) {
                Log.i("MainActivity", "---onNext()--->");
            }

            @Override
            public void onError(Throwable e) {
                Log.i("MainActivity", "---onError()--->" + e.toString());
            }

            @Override
            public void onComplete() {
                Log.i("MainActivity", "---onComplete()--->");
            }
        });

    }

    private void empty() {
        Observable.empty().subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i("MainActivity", "---onSubscribe()--->");
            }

            @Override
            public void onNext(Object o) {
                Log.i("MainActivity", "---onNext()--->");
            }

            @Override
            public void onError(Throwable e) {
                Log.i("MainActivity", "---onError()--->");
            }

            @Override
            public void onComplete() {
                Log.i("MainActivity", "---onComplete()--->");
            }
        });

    }

    private void repeat() {
        Observable.just(1, 2).repeat(3).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i("Observer", "---accept()--->" + integer);
            }
        });
    }

    private void range() {
        Observable.range(10, 5).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i("Observer", "---accept()--->" + integer);
            }
        });
    }

    private void window() {
        Observable.interval(1, TimeUnit.SECONDS) // 间隔一秒发一次
                .take(15) // 最多接收15个
                .window(3, TimeUnit.SECONDS).subscribeOn(Schedulers.io()).observeOn
                (AndroidSchedulers.mainThread()).subscribe(new Consumer<Observable<Long>>() {
            @Override
            public void accept(Observable<Long> longObservable) throws Exception {
                Log.i("Observer", "---SubWindow--->");
                longObservable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers
                        .mainThread()).subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Log.i("Observer", "---accept()--->" + aLong);
                    }
                });
            }
        });

    }

    private void scan() {
        Observable.just(1, 2, 3).scan(new BiFunction<Integer, Integer, Integer>() {
            @Override
            public Integer apply(Integer integer, Integer integer2) throws Exception {
                return integer + integer2;
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i("Observer", "---accept()--->" + integer);
            }
        });
    }

    private void reduce() {
        Observable.just(1, 2, 3).reduce(6, new BiFunction<Integer, Integer, Integer>() {
            @Override
            public Integer apply(Integer integer, Integer integer2) throws Exception {
                return integer + integer2;
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i("Observer", "---accept()--->" + integer);
            }
        });
    }

    private void merge() {
        ArrayList<Integer> integerArrayList = new ArrayList<>();
        for (int i = 10; i < 20; i++) {
            integerArrayList.add(i);
        }
        Observable.merge(Observable.fromIterable(integerArrayList), Observable.just(1, 2, 3, 4,
                5, 6, 7, 8, 9)).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i("Observer", "---accept()--->" + integer);
            }
        });
    }

    private void last() {
        Observable.fromIterable(new ArrayList<String>()).last("123").subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.i("Observer", "---accept()--->" + s);
            }
        });
        Observable.just(1, 2, 3).last(6).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i("Observer", "---accept()--->" + integer);
            }
        });
    }

    private void defer() {
        final Observable<Integer> deferObservable = Observable.defer(new Callable<ObservableSource<Integer>>() {
            @Override
            public ObservableSource<Integer> call() throws Exception {
                return Observable.just(1, 2, 3);
            }
        });
        Log.i("MainActivity", "---defer()--->" + deferObservable);
        deferObservable.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i("Observer", "---onSubscribe()--->" + deferObservable);
            }

            @Override
            public void onNext(Integer integer) {
                Log.i("Observer", "---onNext()--->" + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.i("Observer", "---onError()--->");
            }

            @Override
            public void onComplete() {
                Log.i("Observer", "---onComplete()--->" + deferObservable);
            }
        });
    }

    private void debounce() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                Thread.sleep(505);
                e.onNext(2);//skip
                Thread.sleep(466);
                e.onNext(3);
                Thread.sleep(507);
                e.onNext(4);//skip
                Thread.sleep(400);
                e.onNext(5);
            }
        }).debounce(500, TimeUnit.MILLISECONDS).subscribeOn(Schedulers.newThread()).observeOn
                (AndroidSchedulers.mainThread()).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i("Observer", "---accept()--->" + integer);
            }
        });
    }

    private void single() {
        Single.just(1).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i("Observer", "---accept()--->" + integer);
            }
        });
    }

    private void take() {
        Observable.just(1, 2, 3, 4, 5, 6).take(2).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i("Observer", "---accept()--->" + integer);
            }
        });
    }

    private void skip() {
        Observable.just(1, 2, 3, 4, 5, 6).skip(2).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i("Observer", "---accept()--->" + integer);
            }
        });
    }

    private void doOnNext() {
        Observable.just(1, 2, 3).doOnNext(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i("doOnNext", "---accept()--->做点事情" + integer);
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i("Observer", "---accept()--->" + integer);
            }
        });
    }

    private void interval() {
        Log.i("MainActivity", "---timer()--->" + System.currentTimeMillis());
        mDisposable = Observable.interval(3, 2, TimeUnit.SECONDS).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<Long>() {
            @Override
            public void accept(Long aLong) throws Exception {
                Log.i("observer", aLong + "---accept()--->" + System.currentTimeMillis());
            }
        });
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
        if (mDisposable != null && !mDisposable.isDisposed()) {
            mDisposable.dispose();
        }
        if (mDispos != null && !mDispos.isDisposed()) {
            mDispos.dispose();
        }
    }

    private void timer() {
        Log.i("MainActivity", "---timer()--->" + System.currentTimeMillis());
        Observable.timer(2, TimeUnit.SECONDS).subscribeOn(Schedulers.io()).observeOn
                (AndroidSchedulers.mainThread()).subscribe(new Consumer<Long>() {
            @Override
            public void accept(Long aLong) throws Exception {
                Log.i("observer", aLong + "---accept()--->" + System.currentTimeMillis());
            }
        });
    }

    private void buffer() {
        Observable.just(1, 2, 3, 4, 5).buffer(3, 2).subscribe(new Consumer<List<Integer>>() {
            private int mInt;

            @Override
            public void accept(List<Integer> integers) throws Exception {
                Log.i("Observer", "---accept()--->times" + ++mInt);
                for (Integer item : integers) {
                    Log.i("Observer", "---accept()--->" + item);
                }
            }
        });
    }

    private void filter() {
        Observable.just(1, 2, 3, 4, 5, 6).filter(new Predicate<Integer>() {
            @Override
            public boolean test(Integer integer) throws Exception {
                return integer >= 3;
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i("Observer", "---accept()--->" + integer);
            }
        });
    }

    private void distinct() {
        People people1 = new People();
        People people2 = new People();
        Observable.just(people1, people1, people2, people2).subscribe(new Consumer<People>() {
            @Override
            public void accept(People people) throws Exception {
                Log.i("MainActivity", "---accept()--->" + people);
            }
        });
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onNext(2);
            }
        }).distinct().subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i("Observer", "---accept()--->" + integer);
            }
        });

    }

    private void concatMap() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                Log.i("Observable", "---subscribe()--->1");
                e.onNext(2);
                Log.i("Observable", "---subscribe()--->2");
                e.onNext(3);
                Log.i("Observable", "---subscribe()--->3");
                e.onNext(4);
                Log.i("Observable", "---subscribe()--->4");
                e.onNext(5);
                Log.i("Observable", "---subscribe()--->5");
                e.onNext(6);
                Log.i("Observable", "---subscribe()--->6");
                e.onNext(7);
                Log.i("Observable", "---subscribe()--->7");
                e.onNext(8);
                Log.i("Observable", "---subscribe()--->8");
                e.onNext(9);
                Log.i("Observable", "---subscribe()--->9");
            }
        }).concatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                Log.i("ConcatMap", "---apply()--->" + integer + "---Thread--->" + Thread
                        .currentThread().getName());
                return Observable.just(String.valueOf(integer)).delay(10000, TimeUnit.MILLISECONDS);
            }
        }).subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.i("observer", "---accept()--->" + s);
            }
        });
    }

    private void flatMap() {
        ArrayList<Integer> arrayList = new ArrayList<>();
        for (int i = 1; i < 10; i++) {
            arrayList.add(i);
        }
        Observable.fromIterable(arrayList).flatMap(new Function<Integer,
                ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                Log.i("FlatMap", "---apply()--->" + integer + "---Thread--->" + Thread
                        .currentThread().getName());
                return Observable.just(String.valueOf(integer)).delay(10000, TimeUnit.MILLISECONDS);
            }
        }).subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.i("observer", "---accept()--->" + s);
            }
        });
    }

    private void changeThread() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                String name = Thread.currentThread().getName();
                Log.i("Observable", "---subscribe()--->" + 1 + "---Thread--->" + name);
                e.onNext(2);
                Log.i("Observable", "---subscribe()--->" + 2 + "---Thread--->" + name);
                e.onNext(3);
                Log.i("Observable", "---subscribe()--->" + 3 + "---Thread--->" + name);
                e.onNext(4);
                Log.i("Observable", "---subscribe()--->" + 4 + "---Thread--->" + name);
            }
        }).subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread()).doOnNext
                (new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i("Observer", "---mainThread Accept()--->" + integer + "---Thread--->" +
                        Thread.currentThread().getName());
            }
        }).observeOn(Schedulers.newThread()).doOnNext(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i("Observer", "---newThread Accept()--->" + integer + "---Thread--->" +
                        Thread.currentThread().getName());
            }
        }).observeOn(Schedulers.io()).doOnNext(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i("Observer", "---io Accept()--->" + integer + "---Thread--->" + Thread
                        .currentThread().getName());
            }
        }).observeOn(Schedulers.computation()).doOnNext(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i("Observer", "---computation Accept()--->" + integer + "---Thread--->" +
                        Thread.currentThread().getName());
            }
        }).subscribe();


        Observable.just(1, 2, 3, 4, 5, 6).subscribeOn(Schedulers.io()).observeOn
                (AndroidSchedulers.mainThread()).filter(new Predicate<Integer>() {
            @Override
            public boolean test(Integer integer) throws Exception {
                Log.i("MainActivity", "---test()--->" + Thread.currentThread().getName());
                return integer > 2;
            }
        }).observeOn(Schedulers.io()).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                Log.i("MainActivity", "---apply()--->" + Thread.currentThread().getName());
                return "this is" + integer;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.i("MainActivity", "---accept()--->" + Thread.currentThread().getName() + "   " +
                        "" + "" + "" + "" + "" + "" + "" + "" + "" + "" + "" + "" + "" + "" + ""
                        + "" + "" + "" + s);
            }
        });
    }

    private void concat() {
        Observable.concat(Observable.just(1, 2, 3), Observable.just(4, 5, 6)).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i("Observer", "---accept()--->" + integer);
            }
        });
    }

    private void zip() {
        Observable.zip(createIntegerObservable(), createStringObservable(),
                createBooleanObservable(), new Function3<Integer, String, Boolean, String>() {
            @Override
            public String apply(Integer integer, String s, Boolean b) throws Exception {
                Log.i("Zip", "---apply()--->" + s + integer + b);
                return s + integer + b;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.i("Observer", "---accept()--->" + s);
            }
        });
    }

    private Observable<Boolean> createBooleanObservable() {
        return Observable.create(new ObservableOnSubscribe<Boolean>() {
            @Override
            public void subscribe(ObservableEmitter<Boolean> e) throws Exception {
                e.onNext(true);
                Log.i("BooleanObservable", "---subscribe()--->true");
                e.onNext(true);
                Log.i("BooleanObservable", "---subscribe()--->true");
                e.onNext(true);
                Log.i("BooleanObservable", "---subscribe()--->true");
                e.onNext(true);
                Log.i("BooleanObservable", "---subscribe()--->true");
            }
        });
    }

    private Observable<String> createStringObservable() {
        return Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("A");
                Log.i("StringObservable", "---subscribe()--->A");
                e.onNext("B");
                Log.i("StringObservable", "---subscribe()--->B");
                e.onNext("C");
                Log.i("StringObservable", "---subscribe()--->C");
            }
        });
    }

    private Observable<Integer> createIntegerObservable() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                Log.i("IntegerObservable", "---subscribe()--->1");
                e.onNext(2);
                Log.i("IntegerObservable", "---subscribe()--->2");
                e.onNext(3);
                Log.i("IntegerObservable", "---subscribe()--->3");
                e.onNext(4);
                Log.i("IntegerObservable", "---subscribe()--->4");
                e.onNext(5);
                Log.i("IntegerObservable", "---subscribe()--->5");
            }
        });
    }

    private void map() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                //map()方法促使发送的每一个事件都按照指定的函数去变化
                //Function<T, R>    T为事件变化前的类型,R为事件变化后的类型
                return "this is result " + integer;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.i("observer", "---accept()--->" + s);
            }
        });
    }

    private void baseMethod() {
        //Observable(被观察者、发射器、上游事件)
        //Observable.create()创建Observable对象
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                Log.i("observable", "---subscribe()发送数据--->1");
                // 发送数据
                e.onNext(1);
                Log.i("observable", "---subscribe()发送数据--->2");
                e.onNext(2);
                Log.i("observable", "---subscribe()发送数据--->3");
                e.onNext(3);
                Log.i("observable", "---subscribe()发送数据--->4");
                e.onNext(4);
                // onComplete()之后observer无法接收数据
                //e.onComplete();
                Log.i("observable", "---subscribe()发送数据--->5");
                e.onNext(5);
            }
        });
        observable.subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread());
        //Observer(观察者、接收器、下游事件)
        Observer<Integer> observer = new Observer<Integer>() {
            private int i;
            private Disposable mDisposable;

            @Override
            public void onSubscribe(Disposable d) {
                Log.i("observer", "---onSubscribe()--->订阅" + Thread.currentThread().getName());
                mDisposable = d;
            }

            @Override
            public void onNext(Integer integer) {
                Log.i("observer", "---onNext()接收数据--->" + integer);
                Log.i("observer", "---onNext()接收数据isDisposed--->" + mDisposable.isDisposed());
                i++;
                if (i == 8) {
                    // 在RxJava2.x中,新增的Disposable可以做到切断的操作,让Observer观察者不再接收上游事件
                    mDisposable.dispose();
                    Log.i("observer", "---onNext()接收数据isDisposed--->" + mDisposable.isDisposed());
                }
            }

            @Override
            public void onError(Throwable e) {
                Log.i("observer", "---onError()接收数据发生错误--->");
            }

            @Override
            public void onComplete() {
                Log.i("observer", "---onComplete()接收数据结束--->");
            }
        };
        //observable订阅observer
        observable.subscribe(observer);

    }

    @Override
    protected void onPause() {
        super.onPause();
        mDispos = Observable.timer(1, TimeUnit.SECONDS).subscribe(new Consumer<Long>() {
            @Override
            public void accept(Long aLong) throws Exception {
                Toast.makeText(MainActivity.this, "text" + aLong, Toast.LENGTH_SHORT).show();
            }
        });
    }
}
